package me.shiqiang.im.listeners;

import com.fasterxml.jackson.core.JsonProcessingException;
import me.shiqiang.im.enums.MessageCmd;
import me.shiqiang.im.msg.MsgBody;
import me.shiqiang.im.msg.NewMsg;
import me.shiqiang.im.redis.RedisKey;
import me.shiqiang.im.service.ChatSwitchService;
import me.shiqiang.im.service.MsgService;
import me.shiqiang.im.service.UserClient;
import me.shiqiang.im.utils.JsonUtils;
import me.shiqiang.im.util.RedisPipelineProcessor;
import io.socket.client.Socket;
import org.json.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import java.util.concurrent.ConcurrentLinkedQueue;

/**
 * Created by maliangxue on 2018/4/15.
 */
@Component("P2pMsgListener")
public class P2pMsgListener extends RouterServerListener {
    private Logger logger = LoggerFactory.getLogger(this.getClass());


    @Override
    public String eventName() {
        return "p2pMsg";
    }

    @Autowired
    MsgService service;

    @Autowired
    ChatSwitchService chatSwitchService;

    @Autowired
    ConcurrentLinkedQueue<RedisPipelineProcessor> queue;

    @Autowired
    StringRedisTemplate redisTemplate;

    @Autowired
    Socket socket;

    @Value("${server.id}")
    String serverId;

    @Override
    public void call(Object... objects) {
        JSONObject obj = (JSONObject)objects[0];
        String s = obj.toString();
        try {
            logger.info("p2p收到消息{}", s);
            MsgBody msgBody = JsonUtils.getInstance().fromJson(s, MsgBody.class);
            Long userId = msgBody.getTo();
            if (!chatSwitchService.chatSwitch(userId)) {
                return;
            }
            Long syncKey = msgBody.getSyncKey();
            String serverId = service.getServerId(userId);
            if (!this.serverId.equals(serverId)) {
                logger.debug("消息{}属于服务器[{}],过滤", s, serverId);
                return;
            }

            queue.add(connection -> {
                try {
                    if (redisTemplate.opsForZSet().zCard(RedisKey.clientMsgQueue(userId.toString())) == 0) {
                        connection.zAdd(RedisKey.clientMsgQueue(userId.toString()).getBytes(), syncKey, s.getBytes());
                        connection.expire(RedisKey.clientMsgQueue(userId.toString()).getBytes(), 604800);//7天
                    } else {
                        connection.zAdd(RedisKey.clientMsgQueue(userId.toString()).getBytes(), syncKey, s.getBytes());
                    }
                    NewMsg fromNewMsg = new NewMsg(userId, syncKey,msgBody);
                    socket.emit(MessageCmd.NEW_MSG, new JSONObject(JsonUtils.getInstance().toJson(fromNewMsg)));
                } catch (JsonProcessingException e) {
                    logger.error("转换异常", e);
                }

            });
        } catch (Exception e) {
            logger.error(String.format("消息处理失败：%s", s), e);
        }
    }
}
